# -*- coding: utf-8 -*-
"""
Python 企业级装饰器最佳实践示例

本模块演示了Python装饰器的企业级最佳实践，包括:
- 函数装饰器和类装饰器
- 带参数的装饰器
- 装饰器链和组合
- 缓存和性能优化装饰器
- 权限验证和日志装饰器
- 重试和错误处理装饰器
- 类型检查和验证装饰器
"""

import functools
import logging
import time
from datetime import datetime
from typing import Any, Callable, Dict, List, Optional, TypeVar, Union
from dataclasses import dataclass
from enum import Enum, auto
import threading
import weakref

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format='%(levelname)s:%(name)s:%(message)s'
)
logger = logging.getLogger(__name__)

# 类型变量
F = TypeVar('F', bound=Callable[..., Any])

class UserRole(Enum):
    """用户角色枚举"""
    GUEST = auto()
    USER = auto()
    ADMIN = auto()
    SUPER_ADMIN = auto()

@dataclass
class User:
    """用户数据类"""
    username: str
    role: UserRole
    permissions: List[str]
    is_active: bool = True

# 全局用户上下文（简化示例）
current_user: Optional[User] = None

def timing_decorator(func: F) -> F:
    """计时装饰器 - 测量函数执行时间"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start_time = time.perf_counter()
        try:
            result = func(*args, **kwargs)
            return result
        finally:
            end_time = time.perf_counter()
            execution_time = end_time - start_time
            logger.info(f"函数 {func.__name__} 执行时间: {execution_time:.4f} 秒")
    return wrapper

def retry(max_attempts: int = 3, delay: float = 1.0, exceptions: tuple = (Exception,)):
    """重试装饰器 - 在失败时重试函数执行
    
    Args:
        max_attempts: 最大重试次数
        delay: 重试间隔（秒）
        exceptions: 需要重试的异常类型
    """
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            last_exception = None
            
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        logger.warning(
                            f"函数 {func.__name__} 第 {attempt + 1} 次尝试失败: {e}，"
                            f"{delay} 秒后重试"
                        )
                        time.sleep(delay)
                    else:
                        logger.error(
                            f"函数 {func.__name__} 在 {max_attempts} 次尝试后仍然失败"
                        )
            
            raise last_exception
        return wrapper
    return decorator

def cache_with_ttl(ttl_seconds: int = 300):
    """带TTL的缓存装饰器
    
    Args:
        ttl_seconds: 缓存生存时间（秒）
    """
    def decorator(func: F) -> F:
        cache: Dict[str, tuple] = {}  # key -> (result, timestamp)
        lock = threading.RLock()
        
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 创建缓存键
            cache_key = f"{func.__name__}:{hash((args, tuple(sorted(kwargs.items()))))}"
            current_time = time.time()
            
            with lock:
                # 检查缓存
                if cache_key in cache:
                    result, timestamp = cache[cache_key]
                    if current_time - timestamp < ttl_seconds:
                        logger.debug(f"缓存命中: {func.__name__}")
                        return result
                    else:
                        # 缓存过期，删除
                        del cache[cache_key]
                
                # 执行函数并缓存结果
                result = func(*args, **kwargs)
                cache[cache_key] = (result, current_time)
                logger.debug(f"缓存更新: {func.__name__}")
                return result
        
        # 添加缓存管理方法
        wrapper.clear_cache = lambda: cache.clear()
        wrapper.cache_info = lambda: {
            'size': len(cache),
            'keys': list(cache.keys())
        }
        
        return wrapper
    return decorator

def require_permission(required_permission: str):
    """权限验证装饰器
    
    Args:
        required_permission: 所需权限
    """
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            if current_user is None:
                raise PermissionError("用户未登录")
            
            if not current_user.is_active:
                raise PermissionError("用户账户已被禁用")
            
            if required_permission not in current_user.permissions:
                raise PermissionError(
                    f"用户 {current_user.username} 缺少权限: {required_permission}"
                )
            
            logger.info(
                f"用户 {current_user.username} 执行操作: {func.__name__}"
            )
            return func(*args, **kwargs)
        return wrapper
    return decorator

def validate_types(**type_hints):
    """类型验证装饰器
    
    Args:
        **type_hints: 参数名到类型的映射
    """
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            # 获取函数签名
            import inspect
            sig = inspect.signature(func)
            bound_args = sig.bind(*args, **kwargs)
            bound_args.apply_defaults()
            
            # 验证类型
            for param_name, expected_type in type_hints.items():
                if param_name in bound_args.arguments:
                    value = bound_args.arguments[param_name]
                    if not isinstance(value, expected_type):
                        raise TypeError(
                            f"参数 {param_name} 期望类型 {expected_type.__name__}，"
                            f"实际类型 {type(value).__name__}"
                        )
            
            return func(*args, **kwargs)
        return wrapper
    return decorator

def audit_log(operation: str, sensitive: bool = False):
    """审计日志装饰器
    
    Args:
        operation: 操作描述
        sensitive: 是否为敏感操作
    """
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            start_time = datetime.now()
            user_info = current_user.username if current_user else "匿名用户"
            
            # 记录操作开始
            log_level = logging.WARNING if sensitive else logging.INFO
            logger.log(
                log_level,
                f"[审计] 用户 {user_info} 开始执行 {operation} (函数: {func.__name__})"
            )
            
            try:
                result = func(*args, **kwargs)
                
                # 记录操作成功
                end_time = datetime.now()
                duration = (end_time - start_time).total_seconds()
                logger.log(
                    log_level,
                    f"[审计] 用户 {user_info} 成功完成 {operation}，耗时 {duration:.3f} 秒"
                )
                
                return result
                
            except Exception as e:
                # 记录操作失败
                end_time = datetime.now()
                duration = (end_time - start_time).total_seconds()
                logger.error(
                    f"[审计] 用户 {user_info} 执行 {operation} 失败: {e}，耗时 {duration:.3f} 秒"
                )
                raise
        return wrapper
    return decorator

class RateLimiter:
    """速率限制装饰器类"""
    
    def __init__(self, max_calls: int, time_window: int):
        """
        Args:
            max_calls: 时间窗口内最大调用次数
            time_window: 时间窗口（秒）
        """
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls: Dict[str, List[float]] = {}
        self.lock = threading.RLock()
    
    def __call__(self, func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            user_key = current_user.username if current_user else "anonymous"
            current_time = time.time()
            
            with self.lock:
                # 初始化用户调用记录
                if user_key not in self.calls:
                    self.calls[user_key] = []
                
                # 清理过期的调用记录
                self.calls[user_key] = [
                    call_time for call_time in self.calls[user_key]
                    if current_time - call_time < self.time_window
                ]
                
                # 检查是否超过限制
                if len(self.calls[user_key]) >= self.max_calls:
                    raise RuntimeError(
                        f"速率限制: 用户 {user_key} 在 {self.time_window} 秒内"
                        f"最多只能调用 {self.max_calls} 次"
                    )
                
                # 记录本次调用
                self.calls[user_key].append(current_time)
            
            return func(*args, **kwargs)
        return wrapper

def singleton_decorator(cls):
    """单例装饰器 - 将类转换为单例"""
    instances = {}
    lock = threading.Lock()
    
    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            with lock:
                if cls not in instances:
                    instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    
    return get_instance

def deprecated(reason: str = "", version: str = ""):
    """废弃警告装饰器
    
    Args:
        reason: 废弃原因
        version: 废弃版本
    """
    def decorator(func: F) -> F:
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            import warnings
            message = f"函数 {func.__name__} 已废弃"
            if version:
                message += f" (自版本 {version})"
            if reason:
                message += f": {reason}"
            
            warnings.warn(message, DeprecationWarning, stacklevel=2)
            return func(*args, **kwargs)
        return wrapper
    return decorator

# 示例函数
@timing_decorator
@cache_with_ttl(ttl_seconds=60)
def expensive_calculation(n: int) -> int:
    """模拟耗时计算"""
    time.sleep(0.1)  # 模拟计算时间
    return sum(i * i for i in range(n))

@retry(max_attempts=3, delay=0.5, exceptions=(ValueError, RuntimeError))
@validate_types(value=int, threshold=float)
def unreliable_operation(value: int, threshold: float = 0.5) -> str:
    """模拟不稳定的操作"""
    import random
    if random.random() < threshold:
        raise ValueError(f"随机失败: {value}")
    return f"操作成功: {value}"

@require_permission("admin_access")
@audit_log("管理员操作", sensitive=True)
def admin_operation(data: str) -> str:
    """管理员专用操作"""
    return f"管理员处理数据: {data}"

@RateLimiter(max_calls=5, time_window=10)
def api_endpoint(request_data: str) -> str:
    """API端点示例"""
    return f"处理请求: {request_data}"

@singleton_decorator
class DatabaseConnection:
    """数据库连接单例"""
    
    def __init__(self):
        self.connection_id = id(self)
        logger.info(f"创建数据库连接: {self.connection_id}")
    
    def query(self, sql: str) -> str:
        return f"执行查询 [{self.connection_id}]: {sql}"

@deprecated(reason="请使用 new_function 替代", version="2.0")
def old_function(x: int) -> int:
    """旧版本函数"""
    return x * 2

def demonstrate_basic_decorators() -> None:
    """演示基础装饰器"""
    print("\n======== 1. 基础装饰器 ========")
    
    # 计时装饰器
    result = expensive_calculation(100)
    print(f"✓ 计算结果: {result}")
    
    # 缓存测试
    result2 = expensive_calculation(100)  # 应该从缓存获取
    print(f"✓ 缓存结果: {result2}")
    
    # 缓存信息
    cache_info = expensive_calculation.cache_info()
    print(f"✓ 缓存信息: {cache_info}")

def demonstrate_retry_decorator() -> None:
    """演示重试装饰器"""
    print("\n======== 2. 重试装饰器 ========")
    
    try:
        result = unreliable_operation(42, threshold=0.3)  # 70% 成功率
        print(f"✓ 操作成功: {result}")
    except ValueError as e:
        print(f"✗ 操作最终失败: {e}")

def demonstrate_permission_decorator() -> None:
    """演示权限装饰器"""
    print("\n======== 3. 权限验证装饰器 ========")
    
    global current_user
    
    # 测试无权限用户
    current_user = User(
        username="regular_user",
        role=UserRole.USER,
        permissions=["read_access"]
    )
    
    try:
        admin_operation("敏感数据")
    except PermissionError as e:
        print(f"✗ 权限不足: {e}")
    
    # 测试管理员用户
    current_user = User(
        username="admin_user",
        role=UserRole.ADMIN,
        permissions=["admin_access", "read_access"]
    )
    
    try:
        result = admin_operation("管理员数据")
        print(f"✓ 管理员操作成功: {result}")
    except PermissionError as e:
        print(f"✗ 管理员操作失败: {e}")

def demonstrate_rate_limiter() -> None:
    """演示速率限制装饰器"""
    print("\n======== 4. 速率限制装饰器 ========")
    
    success_count = 0
    for i in range(7):  # 尝试7次调用（限制是5次）
        try:
            result = api_endpoint(f"请求 {i + 1}")
            success_count += 1
            print(f"✓ {result}")
        except RuntimeError as e:
            print(f"✗ 请求 {i + 1} 被限制: {e}")
    
    print(f"✓ 成功请求数: {success_count}/7")

def demonstrate_singleton_decorator() -> None:
    """演示单例装饰器"""
    print("\n======== 5. 单例装饰器 ========")
    
    # 创建多个"连接"
    conn1 = DatabaseConnection()
    conn2 = DatabaseConnection()
    
    print(f"✓ 连接1 ID: {conn1.connection_id}")
    print(f"✓ 连接2 ID: {conn2.connection_id}")
    print(f"✓ 是否为同一实例: {conn1 is conn2}")
    
    # 使用连接
    result = conn1.query("SELECT * FROM users")
    print(f"✓ 查询结果: {result}")

def demonstrate_deprecated_decorator() -> None:
    """演示废弃警告装饰器"""
    print("\n======== 6. 废弃警告装饰器 ========")
    
    import warnings
    with warnings.catch_warnings(record=True) as w:
        warnings.simplefilter("always")
        
        result = old_function(5)
        print(f"✓ 旧函数结果: {result}")
        
        if w:
            print(f"✓ 废弃警告: {w[0].message}")

def run_comprehensive_tests() -> None:
    """运行综合测试套件"""
    print("\n======== 7. 综合测试套件 ========")
    
    test_functions = [
        ("基础装饰器测试", demonstrate_basic_decorators),
        ("重试装饰器测试", demonstrate_retry_decorator),
        ("权限装饰器测试", demonstrate_permission_decorator),
        ("速率限制测试", demonstrate_rate_limiter),
        ("单例装饰器测试", demonstrate_singleton_decorator),
        ("废弃警告测试", demonstrate_deprecated_decorator),
    ]
    
    passed = 0
    total = len(test_functions)
    
    for test_name, test_func in test_functions:
        try:
            test_func()
            print(f"✓ {test_name} 通过")
            passed += 1
        except Exception as e:
            print(f"✗ {test_name} 失败: {e}")
            logger.exception(f"测试失败: {test_name}")
    
    print(f"\n测试结果: {passed}/{total} 通过")
    print("所有企业级装饰器最佳实践演示完成!")

if __name__ == "__main__":
    run_comprehensive_tests()